#include "pch.h"
#include "OrderedTask.h"
#include "Threadpool.h"
#include "..\Common\SRWLock.h"

COrderedTask::COrderedTask()
{
	AssertInUIThread();
}

COrderedTask::~COrderedTask()
{
	AssertInUIThread();
	assert(m_mapTaskQueues.empty());
}

bool COrderedTask::Init()
{
	AssertInUIThread();

	return true;
}

void COrderedTask::Exit()
{
	AssertInUIThread();

	CExclusiveLock lock(m_SRWLock);
	auto mapTaskQueues = std::move(m_mapTaskQueues);
	lock.Unlock();

	// Sequentially clean up pending tasks
	for (auto& [key, taskQueue] : mapTaskQueues)
	{
		while (taskQueue.deqTasks.size())
		{
			taskQueue.deqTasks.pop_front();
		}
	}
}

bool COrderedTask::PostTask(const CValueKey& key, ICallbackPtr&& pTask, const std::unique_ptr<CThreadpool>& pThreadpool)
{
	CExclusiveLock lock(m_SRWLock);
	auto& taskQueue = m_mapTaskQueues[key];
	bool bRunning = taskQueue.bRunning;
	taskQueue.bRunning = true;
	taskQueue.deqTasks.emplace_back(std::move(pTask));
	lock.Unlock();

	// Run the task if it is not running
	return bRunning || pThreadpool->PostTask(&COrderedTask::RunTask, this, key);
}

void COrderedTask::RunTask(const CValueKey& key)
{
	AssertInThreadpool();
	SetInOrderedTask(true);

	for (;;)
	{
		CExclusiveLock lock(m_SRWLock);

		auto it = m_mapTaskQueues.find(key);
		if (it == m_mapTaskQueues.end())	// Exiting
			break;

		if (it->second.deqTasks.empty())
		{
			// Remove the queue if there are no tasks
			m_mapTaskQueues.erase(it);
			break;
		}

		auto pTask = std::move(it->second.deqTasks.front());
		it->second.deqTasks.pop_front();
		lock.Unlock();

		pTask->Invoke();
		pTask = nullptr;
	}

	SetInOrderedTask(false);
}
